/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package baiduschPro;

import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.PropertyValue;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.annotation.behavior.ReadsAttribute;
import org.apache.nifi.annotation.behavior.ReadsAttributes;
import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.io.StreamCallback;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.stream.io.BufferedInputStream;
import org.apache.nifi.stream.io.BufferedOutputStream;
import org.apache.nifi.stream.io.ByteArrayOutputStream;

import baidusch.FormatBaiduSchUrl;
import hylanda.common.FileWr;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

@Tags({"single baidu-search;input "})
@CapabilityDescription("baidu search genarate url")
@SeeAlso({})
@ReadsAttributes({@ReadsAttribute(attribute="", description="")})
@WritesAttributes({@WritesAttribute(attribute="", description="")})

public class MyProcessor2 extends AbstractProcessor {

   
	public static final PropertyDescriptor isdebug = new PropertyDescriptor
            .Builder().name("isdebug")
            .displayName("isdebug")
            .description("isdebug,1写文件到制定目录 0不写")
            .required(false)
            .defaultValue("0")
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();
	public static final PropertyDescriptor debugpath = new PropertyDescriptor
            .Builder().name("debugpath")
            .displayName("debugpath")
            .description("debugpath,debug信息写到目录中，只有当isdebug=1时才有用")
            .required(false)
            .defaultValue("0")
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();

    public static final Relationship MY_RELATIONSHIP = new Relationship.Builder()
            .name("gen success")
            .description("success relationship")
            .build();
    

    private List<PropertyDescriptor> descriptors;

    private Set<Relationship> relationships;

    @Override
    protected void init(final ProcessorInitializationContext context) {
        final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
       
        this.descriptors = Collections.unmodifiableList(descriptors);
        descriptors.add(isdebug);
        descriptors.add(debugpath);

        final Set<Relationship> relationships = new HashSet<Relationship>();
        relationships.add(MY_RELATIONSHIP);
        this.relationships = Collections.unmodifiableSet(relationships);
        
        
    }

    @Override
    public Set<Relationship> getRelationships() {
        return this.relationships;
    }

    @Override
    public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
        return descriptors;
    }

    @OnScheduled
    public void onScheduled(final ProcessContext context) {

  //  	System.out.println("onScheduled is clled");
    }

    @Override
    public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
        FlowFile flowFile = session.get();
        if ( flowFile == null ) {
        	
        	System.out.println("flowfile is null");
           return;
        }
       
		//读取FlowFile
        String content = "";
        InputStream in = session.read(flowFile);
        try {
			byte[] by = readStream(in);
			content = new String(by,"UTF-8");
			
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
        
        PropertyValue debugp = context.getProperty("isdebug");
        String isdebug = debugp.getValue();
        
        PropertyValue debugpathp = context.getProperty("debugpath");
        String debugpath = debugpathp.getValue();
        
        if(isdebug.compareToIgnoreCase("1") == 0)
        {
        	SimpleDateFormat df = new SimpleDateFormat("yyyyMMddHHmmss");//设置日期格式
        	String strtime = df.format(new Date());// new Date()为获取当前系统时间
        	int nlen = debugpath.length();
        	char last = debugpath.charAt(nlen-1);
        	if(last =='/')
        		;
        	else
        		debugpath +="/";
        	
        	String newpath = debugpath+strtime;
        	FileWr.saveFile(newpath, content);
        	try {
				Thread.sleep(1001);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
        }
        
       
      //写flowfile
        String writestr= content;
	     flowFile= session.write(flowFile, new OutputStreamCallback(){
	    		@Override
	    		public void process(OutputStream out) throws IOException{
	    			out.write(writestr.getBytes("UTF-8"));
	    		}
	    	});
	     
	     session.transfer(flowFile,MY_RELATIONSHIP);
	/*	
        
       
		*/
        
    }
     
    public static byte[] readStream(InputStream inStream) throws Exception {  
        ByteArrayOutputStream outSteam = new ByteArrayOutputStream();  
        byte[] buffer = new byte[1024];  
        int len = -1;  
        while ((len = inStream.read(buffer)) != -1) {  
            outSteam.write(buffer, 0, len);  
        }  
        outSteam.close();  
        inStream.close();  
        return outSteam.toByteArray();  
    }  
}

/*
 *  flowFile = session.write(flowFile, new StreamCallback(){
 			@Override
 			public void process(InputStream in,OutputStream out) throws IOException{
 				try(final InputStream inrow = new BufferedInputStream(in);
 				final OutputStream outrow = new BufferedOutputStream(out)){
 				System.out.println(inrow);
 				try {
					byte[] by = readStream(inrow);
					String strse = new String(by);
					System.out.println(strse);
				} catch (Exception e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
					System.out.println("err===");
					out.write("err===".getBytes());
					//session.transfer(flowFile,MY_RELATIONSHIP1);
				}
 				out.write(tmp.getBytes());
 				}
 			}
 			
 		});
 * */
